package com.wing.consumer;

import com.alibaba.fastjson.JSONObject;
import com.tencentcloudapi.tdmq.v20200217.models.CreateSubscriptionRequest;
import com.tencentcloudapi.tdmq.v20200217.models.CreateSubscriptionResponse;
import com.wing.common.exception.BusinessException;
import com.wing.common.utils.JsonResult;
import com.wing.config.BaseDataInit;
import com.wing.config.PulsarClientInit;
import com.wing.order.service.CabInventoryProductService;
import com.wing.redis.utils.redis.RedisHelperImpl;
import com.wing.user.feign.UserClient;
import com.wing.user.model.request.ApplicationAuthForm;
import com.wing.user.model.response.ApplicationAuthVO;
import com.wing.util.PulsarCommons;
import com.wing.util.TencentCloudApiUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

/**
 * 企业应用访问控制 授权下发 (消费端)
 */
@Slf4j
@Component
public class ApplicationAccessConsumer {

    @Value("${pulsar.client.subscriptionName}")
    private String subscriptionName;

    @Value("${pulsar.client.topicUrl}")
    private String topicUrl;
    @Value("${pulsar.client.namespace}")
    private String namespace;
    @Value("${pulsar.client.clusterId}")
    private String clusterId;

    @Autowired
    private UserClient userClient;

    @Autowired
    private BaseDataInit baseDataInit;
    @Autowired
    private CabInventoryProductService cabInventoryProductService;

    @Autowired
    private RedisHelperImpl redisHelper;

    @Autowired
    private TencentCloudApiUtil tencentCloudApiUtil;

    static PulsarClientInit pulsarclientinitultimate =null;
    public ApplicationAccessConsumer(PulsarClientInit pulsarclientinitultimate) {
        ApplicationAccessConsumer.pulsarclientinitultimate = pulsarclientinitultimate;
    }

    @PostConstruct
    @SuppressWarnings("unchecked")
    public void init(){
        //DynamicDataSource.name.set(baseDataInit.ENTERPRISE_DATA.getCode()); // 切换数据源
        String subscriptionName = baseDataInit.ENTERPRISE_DATA.getCode()+"_"+PulsarCommons.APPLICATION_ACCESS_AUTH_SYNS;
        CreateSubscriptionRequest req = new CreateSubscriptionRequest();
        req.setEnvironmentId("company_" + baseDataInit.ENTERPRISE_DATA.getCode());
        req.setTopicName(PulsarCommons.APPLICATION_ACCESS_AUTH_SYNS);
        req.setSubscriptionName(subscriptionName);
        req.setIsIdempotent(true);
        req.setClusterId(clusterId.substring(0,clusterId.length()-1));
        req.setAutoCreatePolicyTopic(true);
        CreateSubscriptionResponse resp = tencentCloudApiUtil.createSubscription(req);
        if(null == resp || !resp.getResult()){
            log.error("创建订阅关系失败" );
            return;
        }
        PulsarClient client = pulsarclientinitultimate.getPulsars();
        try {
            String url = topicUrl.split("/")[0] + "/company_" + baseDataInit.ENTERPRISE_DATA.getCode()+"/";
            Consumer<byte[]> subscribe = client.newConsumer()
                    .subscriptionName(subscriptionName)
                    .topic(url+PulsarCommons.APPLICATION_ACCESS_AUTH_SYNS)
                    .consumerName(subscriptionName)
                    // 配置从最早开始消费，+6+  k否则可能会消费不到历史消息
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                    // 设置消息处理失败后，60秒后再此收到消息。
                    .enableRetry(true)
                    .negativeAckRedeliveryDelay(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)//仅共享消费模式支持重试和死信
                    .deadLetterPolicy(DeadLetterPolicy.builder()
                            //1.当消息消费异常时,消息保存到重试队列,进行重试;
                            //2.最大重试6次后，进入到死信队列,可对死信队列单独进行订阅处理
                            //指定最大重试次数：
                            .maxRedeliverCount(6)
                            //指定重试队列
                            .retryLetterTopic(url+subscriptionName+"-RETRY")
                            //指定死信队列
                            .deadLetterTopic(url+subscriptionName+"-DLQ")
                            .build())
                    .messageListener((MessageListener<byte[]>) (consumer, message) -> {
                        log.info("consumer = " + message.getTopicName());
                        log.info("consumer = " +  message.getPublishTime());
                        log.info("consumer = " +  new String(message.getValue()));
                        processLogic(message.getValue());
                        try {
                            consumer.acknowledge(message);
                        } catch (PulsarClientException e) {
                            log.error("下发授权信息消费确认异常:",e);
                        }
                    })
                    .subscribe();

        } catch (Exception e) {
            log.error("下发授权信息消费确认异常:",e);
        }
    }

    public void processLogic(byte[] msg){
        String data = new String(msg);
        data = StringEscapeUtils.unescapeJavaScript(data);
        log.info("授权信息数据下发:" + data);
        JSONObject jsonObject = JSONObject.parseObject(data.substring(1,data.length()-1));
        //更新或添加

        //授权信息写入数据库
        //存入的是加密后的数据,使用前需要先解密
        JsonResult<ApplicationAuthVO> applicationAuthVOJsonResult = userClient.getApplicationAuthByAppId(jsonObject.getString("appId"));
        if(null == applicationAuthVOJsonResult || applicationAuthVOJsonResult.getCode() != 200) {
            throw new BusinessException("查询应用授权信息失败!");
        }
        ApplicationAuthVO applicationAuthVO = applicationAuthVOJsonResult.getData();
        if(null == applicationAuthVO){
            ApplicationAuthForm applicationAuthForm = new ApplicationAuthForm();
            applicationAuthForm.setAppId(jsonObject.getString("appId"));
            applicationAuthForm.setAuthInfo(jsonObject.getString("data"));
            JsonResult addJsonResult = userClient.addApplicationAuth(applicationAuthForm);
            if(null == addJsonResult || addJsonResult.getCode() != 200) {
                throw new BusinessException("添加应用授权信息失败!");
            }
        }else{
            ApplicationAuthForm applicationAuthForm = new ApplicationAuthForm();
            applicationAuthForm.setAppId(jsonObject.getString("appId"));
            applicationAuthForm.setAuthInfo(jsonObject.getString("data"));
            JsonResult updateJsonResult = userClient.updateApplicationAuth(applicationAuthForm);
            if(null == updateJsonResult || updateJsonResult.getCode() != 200) {
                throw new BusinessException("修改应用授权信息失败!");
            }
        }



    }

}
